package com.ruoyi.iot.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.reflections.Reflections;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


import jakarta.annotation.Resource;
import lombok.Data;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Set;
import java.util.concurrent.Executor;

@Configuration
@Data
@Slf4j
public class MqttConfigure {

	@Value("${spring.mqtt.username}")
	private String username;

	@Value("${spring.mqtt.password}")
	private String password;

	@Value("${spring.mqtt.url}")
	private String hostUrl;

	@Resource
	MqttMessageCallBack mqttMessageCallBack;

	@Resource
	MqttMessageListener mqttMessageListener;

	@Bean
	public MqttConnectOptions mqttConnectOptions(){
		MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
		mqttConnectOptions.setUserName(username);
		mqttConnectOptions.setPassword(password.toCharArray());
		mqttConnectOptions.setServerURIs(new String[]{hostUrl});
		mqttConnectOptions.setConnectionTimeout(5);
		mqttConnectOptions.setCleanSession(true);
		mqttConnectOptions.setKeepAliveInterval(10);
		mqttConnectOptions.setAutomaticReconnect(true);
		return mqttConnectOptions;
	}

	@Bean
	public MqttClient mqttClient(MqttConnectOptions mqttConnectOptions)  {
		MemoryPersistence persistence = new MemoryPersistence();
		MqttClient mqttClient = null;
		try {
			mqttClient = new MqttClient(hostUrl,MqttClient.generateClientId(),persistence);
			mqttClient.setCallback(mqttMessageCallBack);
			mqttClient.connect(mqttConnectOptions);
			if(mqttClient.isConnected()) {
				log.info("连接mqtt成功: {}",hostUrl);
			}
			// 创建 Reflections 实例，用于扫描包
			Reflections reflections = new Reflections("com.ruoyi.iot.handler");
			// 获取包下所有带有 MqttSubscribeService 注解的类
			Set<Class<?>> annotatedClasses = reflections.getTypesAnnotatedWith(com.ruoyi.iot.config.MqttSubscribeService.class);
			// 遍历找到的类并输出 topic 值
			for (Class<?> clazz : annotatedClasses) {
				com.ruoyi.iot.config.MqttSubscribeService annotation = clazz.getAnnotation(com.ruoyi.iot.config.MqttSubscribeService.class);
				if (annotation != null) {
					mqttClient.subscribe(annotation.topic(), mqttMessageListener);
					log.info("订阅topic: {}",annotation.topic());
				}
			}
		} catch (MqttException e) {
			throw new RuntimeException(e);
		}
		return mqttClient;
	}

	@Bean(name = "mqttTaskExecutor")
	public Executor taskExecutor() {
		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		executor.setMaxPoolSize(200);
		executor.setCorePoolSize(50);
		executor.setQueueCapacity(1000);
		executor.setKeepAliveSeconds(300);
		executor.setThreadNamePrefix("mqtt-thread-"); // 线程名称前缀
		executor.initialize();
		return executor;
	}

}
